package spider

import (
	"context"
	"github.com/PuerkitoBio/goquery"
	"log"
	"net/http"
	"net/url"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

type Spider struct {
	ctx            context.Context
	MaxVisitedUrls int       // Максимальное число url, которое посетит паук
	WorkerPoolSize int       // Количество параллельных воркеров
	MaxDepth       int       // Максимальная глубина url
	taskChan       chan Task // Канал для задач на обход url
	mx             *sync.RWMutex
	visitedUrls    map[string]struct{} // Мапа с посещенными url
	wg             *sync.WaitGroup
	wg2            *sync.WaitGroup
	RequestTimeout time.Duration
	workerWait     map[int]bool
	mxWw           *sync.RWMutex
	requests       atomic.Int64
	PageHandler    func(ctx context.Context, link string, depth int, text string)
}

func NewSpider(ctx context.Context) *Spider {
	s := &Spider{
		MaxVisitedUrls: 500,
		WorkerPoolSize: runtime.NumCPU(),
		MaxDepth:       3,
		mx:             &sync.RWMutex{},
		wg:             &sync.WaitGroup{},
		wg2:            &sync.WaitGroup{},
		mxWw:           &sync.RWMutex{},
		ctx:            ctx,
		RequestTimeout: 3 * time.Second,
		PageHandler:    func(ctx context.Context, link string, depth int, text string) {},
	}

	return s
}

func (s *Spider) Start(startUrl string) {

	ctx, cancel := context.WithCancel(s.ctx)
	s.taskChan = make(chan Task, s.WorkerPoolSize)
	s.visitedUrls = make(map[string]struct{})
	s.ctx = ctx
	s.workerWait = make(map[int]bool, s.WorkerPoolSize)

	s.taskChan <- Task{
		url:   startUrl,
		depth: 0,
	}

	defer cancel()

	for i := 0; i < s.WorkerPoolSize; i++ {
		s.wg.Add(1)
		go s.worker(i)
	}

	s.wg.Wait()

	// вычитываем из канала всё что не обработали в отдельно горутине
	go func() {
		for range s.taskChan {
		}
	}()

	// ждём окончания работы тех, кто кладет в канал
	s.wg2.Wait()

	// закрываем канал с заданиями
	close(s.taskChan)
}

func (s *Spider) request(task Task) {
	defer s.requests.Add(-1)

	ctx, cancel := context.WithTimeout(s.ctx, s.RequestTimeout)
	defer cancel()

	req, err := http.NewRequestWithContext(ctx, "GET", task.url, nil)

	if err != nil {
		log.Println("Error http get: " + err.Error())
		return
	}

	resp, err := http.DefaultClient.Do(req)

	if err != nil {
		log.Println("Error http get: " + err.Error())
		return
	}

	defer resp.Body.Close()

	doc, err := goquery.NewDocumentFromReader(resp.Body)
	if err != nil {
		log.Println(err)
		return
	}

	log.Printf("%s : %d : %s", task.url, task.depth, doc.Find("title").First().Text())

	html, err := doc.Html()
	if err != nil {
		log.Println(err)
		return
	}

	s.PageHandler(ctx, task.url, task.depth, html)

	doc.Find("a").Each(func(i int, selection *goquery.Selection) {
		link, exists := selection.Attr("href")
		if !exists {
			log.Println("Not found link in href")
			return
		}

		if task.depth+1 > s.MaxDepth {
			log.Println("MaxDepth reached")
			return
		}

		// make link absolute if it is relative
		base, err := url.Parse(task.url)
		if err != nil {
			log.Println(err)
			return
		}

		absoluteUrl, err := url.Parse(link)
		if err != nil {
			//log.Println("Parse error")
			//log.Println(err)
			return
		}

		absoluteLink := base.ResolveReference(absoluteUrl)
		// remove # fragment from link
		absoluteLink.Fragment = ""

		// кладём в канал в отдельной горутине, чтобы не блокировать воркера
		s.wg2.Add(1)
		go func() {
			defer s.wg2.Done()

			select {
			case <-s.ctx.Done():
				return
			default:
				s.taskChan <- Task{
					url:   absoluteLink.String(),
					depth: task.depth + 1,
				}
			}
		}()
	})
}

func (s *Spider) worker(id int) {
	defer s.wg.Done()

	for {
		select {
		case <-s.ctx.Done():
			return
		case task, ok := <-s.taskChan:

			s.mxWw.Lock()
			s.workerWait[id] = false
			s.mxWw.Unlock()

			if !ok {
				log.Println("Task channel is closed")
				return
			}

			s.mx.RLock()
			countVisitedUrls := len(s.visitedUrls)
			s.mx.RUnlock()

			// если уже посетили больше url чем планировалось - тогда выходим
			if countVisitedUrls >= s.MaxVisitedUrls {
				log.Printf("Reached max visitedUrls: %d", countVisitedUrls)
				return
			}

			// проверим, что мы ещё не были по этому url
			s.mx.RLock()
			_, ok = s.visitedUrls[task.url]
			s.mx.RUnlock()

			if ok {
				log.Printf("Skip visited url: %s", task.url)
				continue
			}

			// отметим этот url в списке посещенных
			s.mx.Lock()
			s.visitedUrls[task.url] = struct{}{}
			s.mx.Unlock()

			s.requests.Add(1)
			s.request(task)
		default:

			s.mxWw.Lock()

			if !s.workerWait[id] {
				s.workerWait[id] = true
			}

			length := len(s.workerWait)
			s.mxWw.Unlock()

			// если ожидающих воркеров столько же сколько всего
			// а также сейчас не выполняется никаких запросов
			// тогда можно выходить
			if s.requests.Load() == 0 && length == s.WorkerPoolSize {
				log.Println("All workers are waiting")
				return
			}

		}

	}
}

type Task struct {
	url   string
	depth int
}
